package dnet

import (
	"dgo/goutils/utils"
	"fmt"
	"net"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

type TcpSpliterRec struct {
	disFlag               int32
	backendwriteflag      int32
	TerminatedFlag        int32
	restartcnt            int
	activeFlag            byte
	owner                 *TcpSpliter
	masterflag            byte
	Frontend_conn         net.Conn
	backend_addr          string
	Backend_conn          net.Conn
	frontend_recvsize     int64
	backend_recvsize      int64
	fronted_recv_monitor  *utils.MonitorSpeedSizeRec
	backend_recv_monitor  *utils.MonitorSpeedSizeRec
	backend_write_monitor *utils.MonitorSpeedSizeRec
	Tag                   interface{}
	DisReason             string
	sessionid             string
	OnRecvBuf             func(conn net.Conn, buf []byte)
	OnWriteBuf            func(sender *TcpSpliterRec, conn net.Conn, buf []byte, err error)
	OnDoneFunc            func(sender *TcpSpliterRec)
}

type SpliterSession struct {
	SessionID         uint32
	Master            *TcpSpliterRec
	Slave             *TcpSpliterRec
	SlaveReconnectCnt int
}

func (this *SpliterSession) SearchVal(val interface{}) bool {
	str := utils.GetStrValue(val, "")
	if len(str) == 0 {
		return true
	}
	return fmt.Sprintf("%d", this.SessionID) == str || strings.Contains(this.Master.BackendRemoteAddr(), str) || strings.Contains(this.Master.FrontendRemoteAddr(), str)
}

func NewTcpSpliterConn(addr string) *TcpSpliterRec {
	rec := &TcpSpliterRec{
		sessionid:             fmt.Sprintf("%s_%d", utils.RandKeyString(12), utils.NewObjectDNA()),
		backend_addr:          addr,
		fronted_recv_monitor:  utils.NewMonitorSpeedSizeRec(),
		backend_write_monitor: utils.NewMonitorSpeedSizeRec(),
		backend_recv_monitor:  utils.NewMonitorSpeedSizeRec(),
	}
	return rec
}

func (this *TcpSpliterRec) String() string {
	return fmt.Sprintf("%s, active:%d(%d)", this.BackendRemoteAddr(), this.activeFlag, this.restartcnt)
}

func (this *TcpSpliterRec) RestartCnt() int {
	return this.restartcnt
}

func (this *TcpSpliterRec) BackendRemoteAddr() string {
	return this.backend_addr
}

func (this *TcpSpliterRec) FrotendRecvInfo() string {
	return this.fronted_recv_monitor.Info()
}

func (this *TcpSpliterRec) BackendWriteInfo() string {
	return this.backend_write_monitor.Info()
}

func (this *TcpSpliterRec) FrontendRemoteAddr() string {
	if this.Frontend_conn != nil {
		return this.Frontend_conn.RemoteAddr().String()
	} else {
		return ""
	}
}

func (this *TcpSpliterRec) IsMaster() bool {
	return this.masterflag == 1
}

func (this *TcpSpliterRec) WriteBackendBuf(buf []byte) (n int, err error) {
	if atomic.CompareAndSwapInt32(&this.backendwriteflag, 0, 1) {
		defer func() {
			this.backendwriteflag = 0
		}()
		conn := this.Backend_conn
		if conn != nil {
			n, err = conn.Write(buf)
			if err != nil {
				this.backend_write_monitor.Delta(int64(n))
			}
			return
		}
		return
	} else {
		this.owner.DoErrFunc(func() error {
			return fmt.Errorf("[%s]WriteBackendBuf重入", this.BackendRemoteAddr())
		})
	}
	return
}

func (this *TcpSpliterRec) OnRecvFrontendBuf(buf []byte) {
	if len(buf) == 0 {
		return
	}

	if this.activeFlag == 0 {
		return
	}

	this.fronted_recv_monitor.DeltaBuf(buf)

	if this.masterflag == 1 {
		return
	}
	this.WriteBackendBuf(buf)

}

func (this *TcpSpliterRec) RequestDisconnect(reason string) {
	if atomic.CompareAndSwapInt32(&this.disFlag, 0, 1) {
		this.DisReason = reason
	}

	if this.masterflag == 1 && this.Frontend_conn != nil {
		this.Frontend_conn.Close()
	}

	if this.Backend_conn != nil {
		this.Backend_conn.Close()
	}
}

/*
*

	重连
*/
func (this *TcpSpliterRec) StartAndReconnect() {
	this.TerminatedFlag = 0

	for this.TerminatedFlag == 0 {
		this.Start() // 重连
		this.restartcnt++
		if !this.IsMaster() {
			time.Sleep(time.Second * 10)
		}
	}
}

func (this *TcpSpliterRec) Start() {
	this.innerConnectBackend()
	if this.masterflag == 1 {
		atomic.AddInt32(&this.owner.MasterOnlineCnt, 1)
	} else {
		atomic.AddInt32(&this.owner.SlaveOnlineCnt, 1)
	}
	defer func() {
		this.activeFlag = 0
		this.owner.onClose(this)
		if this.OnDoneFunc != nil {
			this.OnDoneFunc(this)
		}
	}()
	backend := this.Backend_conn
	frontend := this.Frontend_conn
	if backend == nil || frontend == nil {
		return
	}
	utils.Monitor.AddMonitor(this.fronted_recv_monitor)
	utils.Monitor.AddMonitor(this.backend_write_monitor)
	utils.Monitor.AddMonitor(this.backend_recv_monitor)
	defer func() {
		utils.Monitor.RemoveMonitorRec(this.fronted_recv_monitor)
		utils.Monitor.RemoveMonitorRec(this.backend_write_monitor)
		utils.Monitor.RemoveMonitorRec(this.backend_recv_monitor)
	}()
	this.activeFlag = 1
	wait := new(sync.WaitGroup)
	if this.IsMaster() {
		wait.Add(1)
		// 从前端读取到后端
		go func() {
			defer wait.Done()
			for {
				buf := make([]byte, 1024)
				n, err := frontend.Read(buf)
				if err != nil {
					this.RequestDisconnect(err.Error())
					return
				}
				newBuf := buf[:n]
				if this.OnRecvBuf != nil {
					this.OnRecvBuf(frontend, newBuf)
				}
				if this.owner.OnFrontendRecv != nil {
					this.owner.OnFrontendRecv(this, newBuf)
				}
				this.fronted_recv_monitor.DeltaBuf(newBuf)

				n, err = this.WriteBackendBuf(newBuf)
				if this.OnWriteBuf != nil {
					this.OnWriteBuf(this, backend, buf[:n], err)
				}
				if err != nil {
					this.RequestDisconnect(err.Error())
					return
				}
			}
		}()

	}

	wait.Add(1)
	// 后端读取数据发送到前端
	go DoIoCopyCycle(func() []byte {
		return make([]byte, 1024)
	}, func(buf []byte) (n int, err error) {
		n, err = backend.Read(buf)
		if err != nil {
			this.RequestDisconnect(err.Error())
			return
		}

		newBuf := buf[:n]

		this.backend_recv_monitor.DeltaBuf(newBuf)

		if this.OnRecvBuf != nil {
			this.OnRecvBuf(backend, newBuf)
		}
		if this.owner.OnBackendRecv != nil {
			this.owner.OnBackendRecv(this, newBuf)
		}
		return n, err
	}, func(buf []byte) (n int, err error) {
		if this.IsMaster() {
			n, err = frontend.Write(buf)
			if this.OnWriteBuf != nil {
				this.OnWriteBuf(this, frontend, buf, err)
			}
			if err != nil {
				this.RequestDisconnect(err.Error())
				return
			}
		}
		return
	}, func() {
		backend.Close()
		wait.Done()
	})

	wait.Wait()
}

func (this *TcpSpliterRec) innerConnectBackend() {
	backend, err := net.Dial("tcp", this.backend_addr)
	if err != nil {
		if this.owner.OnErrFunc != nil {
			this.owner.OnErrFunc(err)
		}
		return
	}
	this.Backend_conn = backend
}

/*
*

	将Tcp连接一分为2, master, slave,
	   master 完整转发
	   slave  转发进入的数据到slaveconn, slaveconn的数据不转发到进入的连接
*/
type TcpSpliter struct {
	MasterOnlineCnt     int32
	SlaveOnlineCnt      int32
	terminated          bool
	closeW              *sync.WaitGroup
	frontend            net.Listener
	frontend_addr       string
	backend_master_addr string
	backend_slave_addr  string
	Sessions            utils.GSyncMapGroup
	OnErrFunc           func(err error)
	OnListenSucc        func()
	/**
	阻塞调用:
	返回值:
	 0: 表示没有处理, 继续执行 (默认的连接处理)
	 1: 表示已经处理, 不要继续执行 (需要自己调用: OnProxyConnCb)
	*/
	OnHandleFrontendConnFunc func(frontendconn net.Conn) int

	OnFrontendRecv          func(sender *TcpSpliterRec, buf []byte)
	OnBackendWriteBuf       func(sess *SpliterSession, sender *TcpSpliterRec, buf []byte)
	OnBackendRecv           func(sender *TcpSpliterRec, buf []byte)
	OnBackendRecvBufConvert func(sender *TcpSpliterRec, buf []byte) []byte
	OnDisconnected          func(sender *TcpSpliterRec)
}

func NewTcpSpliter() *TcpSpliter {
	rval := &TcpSpliter{
		closeW:     new(sync.WaitGroup),
		terminated: true,
	}
	return rval
}

func (this *TcpSpliter) RequestDis(sessionid uint32, reason string) bool {
	obj := this.Sessions.Get(sessionid, nil)
	if obj == nil {
		return false
	}

	sess := obj.(*SpliterSession)
	sess.Master.RequestDisconnect(reason)
	return true
}

func (this *TcpSpliter) DoErrFunc(errfunc func() error) {
	if this.OnErrFunc != nil {
		this.OnErrFunc(errfunc())
	}
}

func (this *TcpSpliter) CheckFrontend() net.Listener {
	if this.frontend == nil {
		l, err := net.Listen("tcp", this.frontend_addr)
		if err != nil {
			if this.OnErrFunc != nil {
				this.OnErrFunc(err)
			}
		} else {
			if this.OnListenSucc != nil {
				this.OnListenSucc()
			}
		}
		this.frontend = l
	}
	return this.frontend
}

func (this *TcpSpliter) handleFrontendConn(frontend_conn net.Conn) {
	sess := &SpliterSession{SessionID: utils.NewObjectDNA()}
	this.Sessions.Set(sess.SessionID, sess)
	defer func() {
		this.Sessions.Remove(sess.SessionID)
		frontend_conn.Close()
	}()

	slaveconn := NewTcpSpliterConn(this.backend_slave_addr)
	slaveconn.masterflag = 0
	slaveconn.owner = this
	slaveconn.Frontend_conn = frontend_conn
	slaveconn.OnWriteBuf = func(sender *TcpSpliterRec, conn net.Conn, buf []byte, err error) {
		if this.OnBackendWriteBuf != nil && err != nil && conn == slaveconn.Backend_conn {
			this.OnBackendWriteBuf(sess, sender, buf)
		}
	}

	masterconn := NewTcpSpliterConn(this.backend_master_addr)
	masterconn.masterflag = 1
	masterconn.owner = this
	masterconn.Frontend_conn = frontend_conn
	masterconn.OnWriteBuf = func(sender *TcpSpliterRec, conn net.Conn, buf []byte, err error) {
		if this.OnBackendWriteBuf != nil && err == nil && conn == masterconn.Backend_conn {
			this.OnBackendWriteBuf(sess, sender, buf)
		}
	}
	masterconn.OnRecvBuf = func(conn net.Conn, buf []byte) {
		defer utils.PanicHandlerWithDebugBuf("处理Master接收数据事件", buf)
		if conn == frontend_conn { // 收到前端数据
			slaveconn.OnRecvFrontendBuf(buf)
		}
	}
	masterconn.OnDoneFunc = func(sender *TcpSpliterRec) {
		defer utils.PanicHandler()
		slaveconn.TerminatedFlag = 1
		slaveconn.RequestDisconnect(fmt.Sprintf("master[%s] trans done:%s", sender.backend_addr, sender.DisReason))
	}

	//slaveconn.OnDoneFunc = func(sender *TcpSpliterRec) {
	//	defer utils.PanicHandler()
	//	if this.RequestDisOnSlaveDisconnect {
	//		masterconn.RequestDisconnect(fmt.Sprintf("slave[%s] disconnect", slaveconn.backend_addr))
	//	}
	//}

	sess.Master = masterconn
	sess.Slave = slaveconn

	// 启用重连
	go slaveconn.StartAndReconnect()

	masterconn.Start()

	// 停止重连
	slaveconn.TerminatedFlag = 1

	//
	//
	//
	//
	//this.DoProxyConnect(frontend_conn, this.backend_master_addr, 1, func(conn net.Conn, buf []byte) {
	//
	//}, this.OnProxyConnCb)
}

func (this *TcpSpliter) onClose(rec *TcpSpliterRec) {
	if rec.masterflag == 1 {
		atomic.AddInt32(&this.MasterOnlineCnt, -1)
	} else {
		atomic.AddInt32(&this.SlaveOnlineCnt, -1)
	}
	if this.OnDisconnected != nil {
		this.OnDisconnected(rec)
	}
}

func (this *TcpSpliter) GetBackendAddr() string {
	return this.backend_master_addr
}

func (this *TcpSpliter) GetBackendSlaveAddr() string {
	return this.backend_slave_addr
}

func (this *TcpSpliter) GetFrontendAddr() string {
	return this.frontend_addr
}

func (this *TcpSpliter) Close() {
	this.terminated = true
	listener := this.frontend
	if listener != nil {
		listener.Close()
	}
}

func (this *TcpSpliter) GetActive() bool {
	return !this.terminated
}

func (this *TcpSpliter) BreakConnect() {
	listener := this.frontend
	if listener != nil {
		listener.Close()
	}
}

func (this *TcpSpliter) Open() {
	this.terminated = false
	this.closeW.Add(1)
	defer this.closeW.Done()
	for !this.terminated {
		listener := this.CheckFrontend()
		if listener == nil {
			time.Sleep(time.Second * 10)
			continue
		}
		client, err := listener.Accept()
		if err != nil {
			if this.OnErrFunc != nil {
				this.OnErrFunc(err)
			}
			this.frontend = nil
			continue
		}

		if this.OnHandleFrontendConnFunc != nil {
			if this.OnHandleFrontendConnFunc(client) == 0 {
				go this.handleFrontendConn(client)
			}
		} else {
			go this.handleFrontendConn(client)
		}
	}
}

func (this *TcpSpliter) SetFrontendAddr(val string) bool {
	if this.frontend_addr != val {
		this.frontend_addr = val
		return true
	}
	return false
}

func (this *TcpSpliter) SetBackendMasterAddr(val string) bool {
	if this.backend_master_addr != val {
		this.backend_master_addr = val
		return true
	}
	return false
}

func (this *TcpSpliter) SetBackendSlaveAddr(val string) bool {
	if this.backend_slave_addr != val {
		this.backend_slave_addr = val
		return true
	}
	return false
}
